package com.levi.transaction;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;

/**
 * 生产者发送事务消息
 * 1.创建事务消息生产者，指定生产者组，指定namesrv
 * 2.添加事务监听器
 * 3.发送half消息之后  调用executeLocalTransaction()方法  执行本地事务，执行本地事务有三种结果
 *          执行成功 返回commit,消息对消费者可见，让消费者进行消费
 *          执行失败  返回rollback，销毁消息 ，消费者执行不了
 *          生产者挂掉或者网络波动超时或者没有给broker发送本地事务执行结果，broker会定时回查本地事务，执行checkLocalTransaction()方法看本地事务是否执行成功，默认回查次数15次
 *
 * 此类执行情况:
 * tagA直接commit，被消费
 * tagB   rollback 消息被删除，不会被消费
 * tagC  unknown  执行回查，回查之后commit，被消费
 * 最终tagA和tagC会被消费
 *
 */
public class Producer {
    public static void main(String[] args) throws Exception{
        //1.创建事务生产者，指定生产者组
        TransactionMQProducer producer = new TransactionMQProducer("group1");
        //2.指定nameserver地址
        producer.setNamesrvAddr("192.168.177.129:9876");
        producer.setSendMsgTimeout(1000000);
        //3.添加事务监听器
        producer.setTransactionListener(new TransactionListener() {
            /**
             * 生产者发送half消息之后，在该方法中执行本地的事务
             *
             * @param msg
             * @param arg
             * @return
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                //如果是tagA的消息,则表示本地事务执行成功，返回COMMIT_MESSAGE
                if (StringUtils.equals("tagA", msg.getTags())) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                    //如果是tagA的消息,则表示本地事务执行失败，返回ROLLBACK_MESSAGE
                } else if (StringUtils.equals("tagB", msg.getTags())) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }else{
                    //其他的返回UNKNOW，之后让broker进行回查本地事务
                    return LocalTransactionState.UNKNOW;
                }
            }

            /**
             * 进行事务状态回查
             * FIXME 实际开发中是要查询本地事务表，根据查询结果返回commit还是rollback
             * @param msg
             * @return
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("消息的tag: "+msg.getTags());
                if (StringUtils.equals("tagC", msg.getTags())) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                }else {
                    return null;
                }

            }
        });
        //4.启动producer
        producer.start();
        //5.创建并发送消息，指定topic，tag，消息体,发送三条消息，分别到Tag A B C
        String[] tags = {"tagA", "tagB", "tagC"};
        for (int i = 0; i <3; i++) {
            Message message = new Message("TransacTion_Topic", tags[i], ("事务消息: " + tags[i]).getBytes(StandardCharsets.UTF_8));
            TransactionSendResult result = producer.sendMessageInTransaction(message, null);
            System.out.println("消息发送成功: "+result.getTransactionId()+" 消息发送状态: "+result.getSendStatus());
        }
    }
}
